#include "redis_user_profile_adapter.h"

#include <map>
#include <boost/lexical_cast.hpp>

#include <thrift/Thrift.h>
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/transport/TBufferTransports.h>

#include "base/logging.h"
// #include "redis_executor/scoped_reply.h"

using namespace std;
using namespace boost;
using namespace base;
using namespace apache::thrift;

namespace imserver {

RedisUserProfileAdapter::RedisUserProfileAdapter(RedisLocator * locator) : redis_exec_(locator) {
}

int RedisUserProfileAdapter::Add(const UserProfile & user, int ttl) {
  const int MAX_BUF_LEN = 512;
  char cmd_buf[MAX_BUF_LEN];
  cmd_buf[MAX_BUF_LEN - 1] = '\0';
  snprintf(cmd_buf, MAX_BUF_LEN - 1, "SETEX im:userprofile:%lld %d %%b", user.qid, ttl);

  // serialize
  shared_ptr<transport::TTransport> transport(new transport::TMemoryBuffer(256));
  shared_ptr<protocol::TProtocol> protocol(new protocol::TBinaryProtocol(transport));
  user.write(protocol.get());

  uint8_t * serialized;
  uint32_t size;
  (dynamic_cast<transport::TMemoryBuffer*>(transport.get()))->getBuffer(&serialized, &size);

  ScopedReply reply(redis_exec_.Execute("", cmd_buf, serialized, size));
  if (!reply) {
    LOG_WARN("add user profile into redis error:" << cmd_buf);
  } else {
    LOG_DEBUG("add user profile into redis ok: SET im:userprofile:" << user.qid << " size=" << size);
  }

  return reply ? 0 : -1;
}

int RedisUserProfileAdapter::BatchAdd(const UserProfileList & users) {
  // TODO: 放到任务队列中！
  int ret = 0;
  for(size_t i = 0; i < users.size(); ++i) {
    if (Add(users[i], 60 * 60 * 24) == 0) {
      ++ret;
    }
  }
  return ret;
}


int RedisUserProfileAdapter::Get(int64_t id, UserProfile * user) {
  const int MAX_BUF_LEN = 512;
  char cmd_buf[MAX_BUF_LEN];
  cmd_buf[MAX_BUF_LEN - 1] = '\0';

  snprintf(cmd_buf, MAX_BUF_LEN - 1, "GET im:userprofile:%lld", id);

  ScopedReply reply(redis_exec_.Execute("", cmd_buf));
  if (reply) {
    if (reply->type == REDIS_REPLY_STRING) {
      // deserialize
      shared_ptr<transport::TTransport> transport(new transport::TMemoryBuffer((uint8_t*)reply->str, reply->len));
      shared_ptr<protocol::TProtocol> protocol(new protocol::TBinaryProtocol(transport));
      
      user->read(protocol.get());
      LOG_DEBUG("Get() id=" << user->qid << " username=" << user->username);
      return 0;
    } else {
      LOG_DEBUG("Get() id=" << id << " not found in redis.");
      return 1;
    }
  }
  LOG_WARN("Get() redis " << cmd_buf << " err.");
  return -1;
}

int RedisUserProfileAdapter::BatchGet(const std::vector<int64_t> & ids, UserProfileMap * users) {
  if (ids.empty()) {
    LOG_WARN("BatchGet qids empty!");
    return 0;
  }
  std::stringstream mget_cmd;
  mget_cmd << "MGET";
  std::vector<int64_t>::const_iterator it = ids.begin();
  for(; it != ids.end(); ++it) {
    mget_cmd << " im:userprofile:" << *it;
  }
  LOG_DEBUG("BatchGet cmd=" << mget_cmd.str());

  ScopedReply reply(redis_exec_.Execute("", mget_cmd.str().c_str()));
  if (reply && reply->type == REDIS_REPLY_ARRAY) {
    LOG_DEBUG(mget_cmd.str() << ", redis reply size :" << reply->elements);
    for(size_t i = 0; i < reply->elements; ++i) {
      redisReply * el = reply->element[i];
      if (el->type == REDIS_REPLY_STRING) {
        // deserialize
        shared_ptr<transport::TTransport> transport(new transport::TMemoryBuffer((uint8_t*)el->str, el->len));
        shared_ptr<protocol::TProtocol> protocol(new protocol::TBinaryProtocol(transport));
        UserProfile user;
        user.read(protocol.get());
        
        users->insert(std::make_pair(user.qid, user));
        LOG_DEBUG("id=" << user.qid << " name=" << user.username);
      } else if (el->type == REDIS_REPLY_NIL) {
        LOG_DEBUG(mget_cmd.str() << " item not exist");
      } else {
        LOG_WARN(mget_cmd.str() << " item bad type");
      }
    }
    LOG_INFO("redis MGET [0]=" << ids[0] << " hit=" << users->size() << "/" << ids.size());
    return 0;
  }
  LOG_ERROR(mget_cmd.str() << " redis error.");
  return -1;
}

}

